package com.atguigu.Sink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.been.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @author wky
 * @create 2021-07-14-18:51
 */
public class Sink_ES {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = senv.socketTextStream("hadoop102", 9998);
        SingleOutputStreamOperator<WaterSensor> map = source.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String s) throws Exception {
                String[] split = s.split(",");
                WaterSensor waterSensor = new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                return waterSensor;
            }
        });
        ArrayList<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("hadoop102", 9200));
        httpHosts.add(new HttpHost("hadoop103", 9200));
        httpHosts.add(new HttpHost("hadoop104", 9200));
  /*      ElasticsearchSinkFunction<WaterSensor> function = new ElasticsearchSinkFunction<WaterSensor>() {
            @Override
            public void process(WaterSensor waterSensor, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                //创建写入请求
                IndexRequest request = Requests.indexRequest()
                        .index("sensor")
                        .type("_doc")
                        .id(waterSensor.getId())
                        .source(JSON.toJSONString(waterSensor), XContentType.JSON);
                requestIndexer.add(request);
            }
        };
        ElasticsearchSink.Builder<WaterSensor> sensorBuilder = new ElasticsearchSink.Builder<>(httpHosts, function);
        //因为读的是无界流，所以需要设置刷写时数据的条数，设置为1代表，来一条往ES中写一条
        sensorBuilder.setBulkFlushMaxActions(1);
        map.addSink(sensorBuilder.build());*/
        ElasticsearchSink.Builder<WaterSensor> sensorBuilder = new ElasticsearchSink.Builder<>(httpHosts, new ElasticsearchSinkFunction<WaterSensor>() {
            @Override
            public void process(WaterSensor element, RuntimeContext ctx, RequestIndexer indexer) {

                //创建Index请求，可以指定要写入的索引，类型，id
                IndexRequest indexRequest = new IndexRequest("sensor-0225", "_doc", element.getId());
                //将数据转为json并声明为json，写入ES
                System.out.println("正在写入......");
                indexRequest.source(JSON.toJSONString(element), XContentType.JSON);
                indexer.add(indexRequest);
            }
        });

        //因为读的是无界流，所以需要设置刷写时数据的条数，设置为1代表，来一条往ES中写一条
        sensorBuilder.setBulkFlushMaxActions(1);
        ElasticsearchSink<WaterSensor> elasticsearchSink = sensorBuilder.build();
        //System.out.println("222");
        map.addSink(elasticsearchSink);
        //System.out.println("333");
        senv.execute();

    }
}
